package com.bw.Test;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.DateFormatUtil;
import com.bw.MyKafkaUtil;
import com.bw.bean.OrderInfo;
import com.bw.fun.AsyncIOFuncykwide;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.phoenix.util.DateUtil;
import ru.yandex.clickhouse.ClickHouseUtil;

import java.text.SimpleDateFormat;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;

public class Test8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(new MyKafkaUtil().getKafkaConsumer("dwd-pay-detail_yk9", "test4"+System.currentTimeMillis()));

        SingleOutputStreamOperator<DwsTradeOrgOrderDayBean> mapDS = stream.map(
                new MapFunction<String, DwsTradeOrgOrderDayBean>() {
                    @Override
                    public DwsTradeOrgOrderDayBean map(String jsonStr) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        JSONObject data = jsonObject.getJSONObject("data");


                        OrderInfo dwdTradeOrderDetailBean = JSON.parseObject(data.toJSONString(), OrderInfo.class);


                        String create_time = data.getString("create_time");
                        //处理时区问题
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                        sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                        long time = sdf.parse(create_time).getTime();
                        dwdTradeOrderDetailBean.setCreate_ts(time);



                        //处理
                        DwsTradeOrgOrderDayBean bean = DwsTradeOrgOrderDayBean.builder()
                                .senderDistrictId(dwdTradeOrderDetailBean.getSender_district_id())
                                .cityId(dwdTradeOrderDetailBean.getSender_city_id())
                                .orderAmountBase(dwdTradeOrderDetailBean.getAmount())
                                .orderCountBase(1L)
                                .ts(dwdTradeOrderDetailBean.getCreate_ts() )
                                .build();
                        return bean;
                    }
                }
        );
        //TODO 4.关联机构维度

        SingleOutputStreamOperator<DwsTradeOrgOrderDayBean> dwsTradeOrgOrderDayBeanSingleOutputStreamOperator = AsyncDataStream.unorderedWait(mapDS, new AsyncIOFuncykwide(), 1000, TimeUnit.MILLISECONDS, 100);


//        dwsTradeOrgOrderDayBeanSingleOutputStreamOperator.print();

        //TODO 5.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<DwsTradeOrgOrderDayBean> withWatermarkDS = dwsTradeOrgOrderDayBeanSingleOutputStreamOperator.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<DwsTradeOrgOrderDayBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsTradeOrgOrderDayBean>() {
                                    @Override
                                    public long extractTimestamp(DwsTradeOrgOrderDayBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );


//        withWatermarkDS.print();

//        //TODO 6.按照机构id进行分组
        KeyedStream<DwsTradeOrgOrderDayBean, String> keyedDS = withWatermarkDS.keyBy(DwsTradeOrgOrderDayBean::getOrgId);

        //TODO 7.开窗
        WindowedStream<DwsTradeOrgOrderDayBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(30)));

        SingleOutputStreamOperator<DwsTradeOrgOrderDayBean> reduceStream = windowDS.reduce(new ReduceFunction<DwsTradeOrgOrderDayBean>() {
            @Override
            public DwsTradeOrgOrderDayBean reduce(DwsTradeOrgOrderDayBean t1, DwsTradeOrgOrderDayBean t2) throws Exception {
                t1.setOrderAmountBase(t1.getOrderAmountBase().add(t2.getOrderAmountBase()));
                t1.setOrderCountBase(t1.getOrderCountBase() + t2.getOrderCountBase());
                return t1;
            }
        }, new WindowFunction<DwsTradeOrgOrderDayBean, DwsTradeOrgOrderDayBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<DwsTradeOrgOrderDayBean> iterable, Collector<DwsTradeOrgOrderDayBean> collector) throws Exception {
                DwsTradeOrgOrderDayBean next = iterable.iterator().next();
                String start = DateFormatUtil.tsToDateTime(timeWindow.getStart());
                next.setCurDate(start);
                next.setTs(System.currentTimeMillis());
                collector.collect(next);
            }
        });


        reduceStream.print();

        env.execute();

//        //TODO 11.将结果写到CK中
//        withCityDS.print(">>>>");
//        withCityDS.addSink(
//                ClickHouseUtil.getJdbcSink("insert into dws_trade_org_order_day_base values(?,?,?,?,?,?,?,?)")
//        );

    }
}
